package com.mini.gateway.core.modules.log.service;

import com.alibaba.fastjson.JSON;
import com.mini.gateway.core.modules.log.entity.ApiLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * ApiLogKafkaListener
 * 监听Api Log topic
 *
 */
@Slf4j
@Component
public class ApiLogKafkaListener {

    @Autowired
    private ApiLogService apiLogService;

    @KafkaListener(topics = ApiLogConstants.KAFKA_TOPIC_APILOG, groupId = ApiLogConstants.KAFKA_GROUP_APILOG)
    public void processMessage(ConsumerRecord<Integer, String> record) {
        try {
            String value = record.value();
            ApiLog apiLog = JSON.parseObject(value, ApiLog.class);
            ApiLog _temp = apiLogService.getBySerialNo(apiLog.getSerialNo());
            if (_temp != null) {
                apiLogService.updateResult(_temp.getId(), apiLog.getResponseCode(), apiLog.getEndTime());
            } else {
                apiLogService.save(apiLog);
            }
        } catch (Exception ex) {
            log.info("【Api Log Kafka process】exception", ex);
        }

    }
}
